package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.PageViewStats;
import com.atguigu.utils.MyClickHouseUtil;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.Tuple5;

import java.text.SimpleDateFormat;
import java.time.Duration;

public class PageViewStats10sApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // ck
//        env.enableCheckpointing(5 * 60000L);
//        env.getCheckpointConfig().setCheckpointTimeout(5 * 60000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000L);
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));
//
//        //
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/210927");


        // 从kafka读取数据
        String sourceTopic = "dwd_page_log";
        String groupId = "page_view_stats_10s_app_2022";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaSource(sourceTopic, groupId));

        // 转换成javaBean
        SingleOutputStreamOperator<PageViewStats> pageViewStatsWithWMDS = kafkaDS.map(line -> {
            //
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");
            JSONObject page = jsonObject.getJSONObject("page");
            String lastPageId = page.getString("last_page_id");

            Long sv_ct = 0L;
            if (lastPageId == null) {
                sv_ct = 1L;
            }

            return new PageViewStats("", "",
                    page.getString("page_id"),
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    1L,
                    sv_ct,
                    page.getLong("during_time"),
                    jsonObject.getLong("ts"));

        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<PageViewStats>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<PageViewStats>() {
                    @Override
                    public long extractTimestamp(PageViewStats element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        // 分组、开窗、聚合
        SingleOutputStreamOperator<PageViewStats> resultDS = pageViewStatsWithWMDS.keyBy(new KeySelector<PageViewStats, Tuple5<String, String, String, String, String>>() {

                    @Override
                    public Tuple5<String, String, String, String, String> getKey(PageViewStats value) throws Exception {
                        return new Tuple5<>(value.getPage_id(),
                                value.getAr(),
                                value.getCh(),
                                value.getVc(),
                                value.getIs_new());
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<PageViewStats>() {
                            @Override
                            public PageViewStats reduce(PageViewStats value1, PageViewStats value2) throws Exception {
                                value1.setPv_ct(value1.getPv_ct() + value2.getPv_ct());
                                value1.setSv_ct(value1.getSv_ct() + value2.getSv_ct());
                                value1.setDur_time(value1.getDur_time() + value2.getDur_time());
                                return value1;
                            }
                        },
                        new WindowFunction<PageViewStats, PageViewStats, Tuple5<String, String, String, String, String>, TimeWindow>() {
                            @Override
                            public void apply(Tuple5<String, String, String, String, String> stringStringStringStringStringTuple5, TimeWindow window, Iterable<PageViewStats> input, Collector<PageViewStats> out) throws Exception {
                                PageViewStats next = input.iterator().next();

                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                next.setStt(sdf.format(window.getStart()));
                                next.setEdt(sdf.format(window.getEnd()));

                                out.collect(next);
                            }
                        });

        resultDS.print("result>>>>>>");
        resultDS.addSink(MyClickHouseUtil.getClickHouseSink("insert into dws_pv_vc_ch_isnew_ar_10s_2022 values(?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
